//package com.we.risk.attribution.resparse;
//
//import com.alibaba.fastjson.JSONObject;
//import com.we.flink.utils.WeKafkaPropertyReader;
//import com.we.risk.attribution.AttributionInfo;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.RichFlatMapFunction;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.api.common.state.MapState;
//import org.apache.flink.api.common.state.MapStateDescriptor;
//import org.apache.flink.api.common.state.StateTtlConfig;
//import org.apache.flink.api.common.time.Time;
//import org.apache.flink.api.java.functions.KeySelector;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.connector.kafka.source.KafkaSource;
//import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
//import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
//import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
//import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
//import org.apache.flink.util.Collector;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import java.io.IOException;
//import java.text.SimpleDateFormat;
//import java.util.*;
//
//import static com.we.risk.attribution.AttributionInfo.*;
//import static com.we.risk.attribution.AttributionInfo.DEVICEID;
//
//public class KfkResParse {
//    public static final String RELEASEPROP =
//            "risk/attribution/click/kfk_click_attribution_test.properties";
//    public static Logger LOG = LoggerFactory.getLogger(KfkResParse.class);
//
//    public static final String DATEFORMAT = "yyyy-MM-dd";
//    public static final String LABELNAME = "label_name";
//    public static final String LABELVALUE = "label_value";
//
//    public static void main(String[] args) throws IOException {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        try {
//            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
//            /** RocksDB */
//            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
//            /** checkpoint configure */
//            CheckpointConfig ckConf = env.getCheckpointConfig();
//            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
//            ckConf.setCheckpointTimeout(60 * 60 * 1000);
//            ckConf.setMaxConcurrentCheckpoints(1);
//            ckConf.setMinPauseBetweenCheckpoints(500);
//            ckConf.enableExternalizedCheckpoints(
//                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//
//            ckConf.enableUnalignedCheckpoints();
//
//            /** Consumer Kafka */
//            String srcKafkaTopic = paramReader.getTmpKfkTopic();
//            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
//            String srckfkGrupId = paramReader.getTmpKfkGroupId();
//
//            Properties consumProp = new Properties();
//            consumProp.setProperty("group.id", srckfkGrupId);
//            consumProp.setProperty(
//                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//            consumProp.setProperty(
//                    "value.deserializer",
//                    "org.apache.kafka.common.serialization.StringDeserializer");
//
//            /** set offset to latest */
//            KafkaSource<String> kfkSource =
//                    KafkaSource.<String>builder()
//                            .setBootstrapServers(srcKafkaBootStrapServer)
//                            .setTopics(srcKafkaTopic)
//                            .setStartingOffsets(OffsetsInitializer.earliest())
//                            .setProperties(consumProp)
//                            .setValueOnlyDeserializer(new SimpleStringSchema())
//                            .build();
//
//            SingleOutputStreamOperator<String> input =
//                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "kfkResParse")
//                            .uid("kfkResParse-source");
//
//            SingleOutputStreamOperator<String> out =
//                    input.flatMap(new attributionResParesFlatMap())
//                            .keyBy(
//                                    new KeySelector<AttributionInfo, String>() {
//                                        @Override
//                                        public String getKey(AttributionInfo value)
//                                                throws Exception {
//                                            return value.getDevice_id();
//                                        }
//                                    })
//                            .flatMap(new KfkResParse.clickAttributionProcess());
//
//            /** sink to Kafka */
//            String sinkkfkTopic = paramReader.getKfkTopic();
//            Properties sinkProp = new Properties();
//            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
//            //            sinkProp.setProperty("max.request.size", String.valueOf(KFKMSGMAXSIZE));
//            int sinkkfkPartitions = paramReader.getKfkPartitions();
//
//            FlinkKafkaProducer<String> kafkaProducer =
//                    new FlinkKafkaProducer<String>(
//                            sinkkfkTopic,
//                            new KfkResParse.WeKafkaKeyedSerializationSchema(),
//                            sinkProp,
//                            Optional.of(new KfkResParse.WeKafkaCustomPartitioner()));
//
//            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);
//
//            env.execute(KfkResParse.class.toString());
//        } catch (Exception e) {
//            LOG.error("Exception: " + e.getMessage());
//            e.printStackTrace();
//        }
//    }
//
//    private static class attributionResParesFlatMap
//            extends RichFlatMapFunction<String, String> {
//
//        @Override
//        public void flatMap(String value, Collector<String> out) throws Exception {
//            try {
//                JSONObject inputJson = JSONObject.parseObject(value);
//                if (inputJson != null) {
//                    Set<String> keySet = inputJson.keySet();
//                    if(keySet != null) {
//                        Iterator<String> iterator = keySet.iterator();
//                        while(iterator.hasNext()) {
//                            String jsonKey = iterator.next();
//                            Object jsonVal = inputJson.get(jsonKey);
//
//                        }
//                    }
//                }
//            } catch (Exception e) {
//                LOG.error("attributionResParesFlatMap: " + e.toString());
//                LOG.error("attributionResParesFlatMap input: " + value);
//            }
//        }
//
//    }
//
//    private static class clickAttributionProcess
//            extends RichFlatMapFunction<AttributionInfo, String> {
//        MapStateDescriptor<Long, JSONObject> mStateDesc;
//        MapState<Long, JSONObject> clkAttributMState;
//        SimpleDateFormat sdf;
//
//        @Override
//        public void open(Configuration parameters) throws Exception {
//            StateTtlConfig ttlConfig =
//                    StateTtlConfig.newBuilder(Time.hours(24)) // dt 按天分区 只存24小时
//                            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
//                            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
//                            .cleanupInRocksdbCompactFilter(100000L)
//                            .build();
//
//            mStateDesc =
//                    new MapStateDescriptor<>("clickAttributionDesc", Long.class, JSONObject.class);
//            mStateDesc.enableTimeToLive(ttlConfig);
//
//            clkAttributMState = getRuntimeContext().getMapState(mStateDesc);
//
//            sdf = new SimpleDateFormat(DATEFORMAT);
//        }
//
//        @Override
//        public void flatMap(AttributionInfo input, Collector<String> out) throws Exception {
//            try {
//                LOG.debug("##input: " + input.toString());
//                String unique_channel = input.getUnique_channel();
//                long receive_time = input.getReceive_time();
//                // dt_time 秒级时间戳
//                long dt_time = getStartTimeOfDay(receive_time, sdf);
//
//                // 分区判断
//                LOG.debug(
//                        "contains dt_time: " + dt_time + " " + clkAttributMState.contains(dt_time));
//                if (clkAttributMState.contains(dt_time)) {
//                    JSONObject stateDtJson = clkAttributMState.get(dt_time);
//                    if (stateDtJson != null) {
//                        if (!stateDtJson.containsKey(unique_channel + 0)) {
//                            // new channel
//                            input.setRecv_cnt(1);
//                            // rt_type0 最早信息
//                            stateDtJson.put(unique_channel + 0, input);
//                            // rt_type1 最晚信息
//                            stateDtJson.put(unique_channel + 1, input);
//                            // 更新MapState
//                            clkAttributMState.put(dt_time, stateDtJson);
//
//                            // 发送下游 需要debug确认
//                            input.setRt_type(0);
//                            LOG.debug("new channel debug0: " + input.toString());
//                            directOutputJson(input, dt_time, out);
//
//                            input.setRt_type(1);
//                            LOG.debug("new channel debug1: " + input.toString());
//                            directOutputJson(input, dt_time, out);
//                        } else {
//                            AttributionInfo earlyInfo =
//                                    (AttributionInfo) stateDtJson.get(unique_channel + 0);
//                            AttributionInfo stateEarlyBak = new AttributionInfo(earlyInfo);
//                            // 有流入cnt就+1
//                            int recvCnt = earlyInfo.getRecv_cnt() + 1;
//                            long earlyTime = earlyInfo.getReceive_time();
//
//                            if (receive_time < earlyTime) {
//                                input.setRecv_cnt(recvCnt);
//                                stateDtJson.put(unique_channel + 0, input);
//                            } else {
//                                earlyInfo.setRecv_cnt(recvCnt);
//                                stateDtJson.put(unique_channel + 0, earlyInfo);
//                            }
//                            AttributionInfo outAttribution =
//                                    (AttributionInfo) stateDtJson.get(unique_channel + 0);
//                            outAttribution.setRt_type(0);
//
//                            // 发送下游 需要debug确认
//                            LOG.debug("old debug0: " + outAttribution.toString());
//                            cmpOutputJson(outAttribution, stateEarlyBak, dt_time, out);
//
//                            AttributionInfo lateInfo =
//                                    (AttributionInfo) stateDtJson.get(unique_channel + 1);
//                            AttributionInfo stateLateBak = new AttributionInfo(lateInfo);
//                            long lateTime = lateInfo.getReceive_time();
//
//                            if (receive_time > lateTime) {
//                                input.setRecv_cnt(recvCnt);
//                                stateDtJson.put(unique_channel + 1, input);
//                            } else {
//                                lateInfo.setRecv_cnt(recvCnt);
//                                stateDtJson.put(unique_channel + 1, lateInfo);
//                            }
//                            AttributionInfo outLateAttribution =
//                                    (AttributionInfo) stateDtJson.get(unique_channel + 1);
//                            outLateAttribution.setRt_type(1);
//                            LOG.debug("old debug1: " + outLateAttribution.toString());
//                            cmpOutputJson(outLateAttribution, stateLateBak, dt_time, out);
//
//                            clkAttributMState.put(dt_time, stateDtJson);
//                        }
//                    }
//                } else {
//                    // new dt
//                    JSONObject dtTimeJson = new JSONObject();
//
//                    input.setRecv_cnt(1);
//                    // rt_type0 最早信息
//                    dtTimeJson.put(unique_channel + 0, input);
//                    // rt_type1 最晚信息
//                    dtTimeJson.put(unique_channel + 1, input);
//                    // 更新MapState
//                    clkAttributMState.put(dt_time, dtTimeJson);
//
//                    // 发送下游 需要debug确认
//                    input.setRt_type(0);
//                    LOG.debug("new dt debug0: " + input.toString());
//                    directOutputJson(input, dt_time, out);
//
//                    input.setRt_type(1);
//                    LOG.debug("new dt debug1: " + input.toString());
//                    directOutputJson(input, dt_time, out);
//                }
//
//            } catch (Exception e) {
//                LOG.error("excep: " + e.toString());
//                LOG.error("excep input: " + input);
//            }
//        }
//
//        private void cmpOutputJson(
//                AttributionInfo value, AttributionInfo state, long dt_time, Collector<String> out) {
//            LOG.debug("@@value:" + value.toString());
//            LOG.debug("@@state:" + state.toString());
//            String device_id = value.getDevice_id();
//            String unique_channel = value.getUnique_channel();
//
//            JSONObject outJson = new JSONObject();
//            // unique key
//            outJson.put(DEVICEID, device_id);
//            outJson.put(DTTIME, dt_time);
//            outJson.put(CHANNEL, unique_channel);
//            outJson.put(RTTYPE, value.getRt_type());
//
//            if (value.getRecv_cnt() != state.getRecv_cnt()) {
//                outJson.put(LABELNAME, RECVCNT);
//                outJson.put(LABELVALUE, value.getRecv_cnt());
//                LOG.debug("#cmpOut: " + outJson.toString());
//                out.collect(outJson.toString());
//            }
//
//            String input_adv_id = value.getHfq_advertiser_id();
//            String st_adv_id = state.getHfq_advertiser_id();
//            LOG.debug("input_adv_id: " + (input_adv_id == null ? "null" : input_adv_id));
//            LOG.debug("st_adv_id: " + (st_adv_id == null ? "null" : st_adv_id));
//            if (input_adv_id != null && st_adv_id != null) {
//                if (!input_adv_id.equals(st_adv_id)) {
//                    outJson.put(LABELNAME, HFQADVSTID);
//                    outJson.put(LABELVALUE, input_adv_id);
//                    LOG.debug("#cmpOut1: " + outJson.toString());
//                    out.collect(outJson.toString());
//                }
//            } else if (input_adv_id != st_adv_id) {
//                outJson.put(LABELNAME, HFQADVSTID);
//                outJson.put(LABELVALUE, input_adv_id);
//                LOG.debug("#cmpOut2: " + outJson.toString());
//                out.collect(outJson.toString());
//            }
//
//            String input_cmpg_id = value.getHfq_campaign_id();
//            String st_cmpg_id = state.getHfq_campaign_id();
//            LOG.debug("input_cmpg_id: " + (input_cmpg_id == null ? "null" : input_cmpg_id));
//            LOG.debug("st_cmpg_id: " + (st_cmpg_id == null ? "null" : st_cmpg_id));
//            if (input_cmpg_id != null && st_cmpg_id != null) {
//                if (!input_cmpg_id.equals(st_cmpg_id)) {
//                    outJson.put(LABELNAME, HFQCMPGID);
//                    outJson.put(LABELVALUE, input_cmpg_id);
//                    LOG.debug("#cmpOut1: " + outJson.toString());
//                    out.collect(outJson.toString());
//                }
//            } else if (input_cmpg_id != st_cmpg_id) {
//                outJson.put(LABELNAME, HFQCMPGID);
//                outJson.put(LABELVALUE, input_cmpg_id);
//                LOG.debug("#cmpOut2: " + outJson.toString());
//                out.collect(outJson.toString());
//            }
//
//            String input_aid = value.getHfq_aid();
//            String st_aid = state.getHfq_aid();
//            LOG.debug("input_aid: " + (input_aid == null ? "null" : input_aid));
//            LOG.debug("st_aid: " + (st_aid == null ? "null" : st_aid));
//            if (input_aid != null && st_aid != null) {
//                if (!input_aid.equals(st_aid)) {
//                    outJson.put(LABELNAME, HFQAID);
//                    outJson.put(LABELVALUE, input_aid);
//                    LOG.debug("#cmpOut1: " + outJson.toString());
//                    out.collect(outJson.toString());
//                }
//            } else if (input_aid != st_aid) {
//                outJson.put(LABELNAME, HFQAID);
//                outJson.put(LABELVALUE, input_aid);
//                LOG.debug("#cmpOut2: " + outJson.toString());
//                out.collect(outJson.toString());
//            }
//
//            String input_clk_id = value.getHfq_click_id();
//            String st_clk_id = state.getHfq_click_id();
//            LOG.debug("input_clk_id: " + (input_clk_id == null ? "null" : input_clk_id));
//            LOG.debug("st_clk_id: " + (st_clk_id == null ? "null" : st_clk_id));
//            if (input_clk_id != null && st_clk_id != null) {
//                if (!input_clk_id.equals(st_clk_id)) {
//                    outJson.put(LABELNAME, HFQCLKID);
//                    outJson.put(LABELVALUE, value.getHfq_click_id());
//                    LOG.debug("#cmpOut1: " + outJson.toString());
//                    out.collect(outJson.toString());
//                }
//            } else if (value.getHfq_click_id() != st_clk_id) {
//                outJson.put(LABELNAME, HFQCLKID);
//                outJson.put(LABELVALUE, value.getHfq_click_id());
//                LOG.debug("#cmpOut2: " + outJson.toString());
//                out.collect(outJson.toString());
//            }
//        }
//
//        private void directOutputJson(AttributionInfo value, long dt_time, Collector<String> out) {
//            String device_id = value.getDevice_id();
//            String unique_channel = value.getUnique_channel();
//
//            JSONObject outJson = new JSONObject();
//            // unique key
//            outJson.put(DEVICEID, device_id);
//            outJson.put(DTTIME, dt_time);
//            outJson.put(CHANNEL, unique_channel);
//            outJson.put(RTTYPE, value.getRt_type());
//
//            outJson.put(LABELNAME, RECVCNT);
//            outJson.put(LABELVALUE, value.getRecv_cnt());
//            //            LOG.debug("#direct: " + outJson.toString());
//            out.collect(outJson.toString());
//
//            outJson.put(LABELNAME, HFQADVSTID);
//            outJson.put(LABELVALUE, value.getHfq_advertiser_id());
//            //            LOG.debug("#direct: " + outJson.toString());
//            out.collect(outJson.toString());
//
//            outJson.put(LABELNAME, HFQCMPGID);
//            outJson.put(LABELVALUE, value.getHfq_campaign_id());
//            //            LOG.debug("#direct: " + outJson.toString());
//            out.collect(outJson.toString());
//
//            outJson.put(LABELNAME, HFQAID);
//            outJson.put(LABELVALUE, value.getHfq_aid());
//            //            LOG.debug("#direct: " + outJson.toString());
//            out.collect(outJson.toString());
//
//            outJson.put(LABELNAME, HFQCLKID);
//            outJson.put(LABELVALUE, value.getHfq_click_id());
//            //            LOG.debug("#direct: " + outJson.toString());
//            out.collect(outJson.toString());
//        }
//    }
//
//    private static long getStartTimeOfDay(long now, SimpleDateFormat sdf) {
//
//        String date = sdf.format(new Date(now));
//        long res = 0;
//        try {
//            res = sdf.parse(date).getTime() / 1000;
//        } catch (Exception e) {
//            LOG.error("sdf parse date error! time: " + now);
//            e.printStackTrace();
//        }
//        return res;
//    }
//
//
//
//    private static String formatIdfaOaid(String input) {
//        String res = null;
//        if (input != null
//                && !"9F89C84A559F573636A47FF8DAED0D33"
//                .equals(input) // '00000000-0000-0000-0000-000000000000'
//                && !"CD9E459EA708A948D5C2F5A6CA8838CF"
//                .equals(input) // '00000000000000000000000000000000'
//                && !"D41D8CD98F00B204E9800998ECF8427E".equals(input) // ''
//                && !"6E0D8DA38B72B0A55A28F0764389A1E7".equals(input) // '__IDFA__'
//                && !"A261C265DF3A6D24B5AD3CC1DDAFD8E6".equals(input) // '__OAID__'
//                && !"A5452BD63942BFCA9CD8E1C09223383E".equals(input) // '__IDFA2__'
//                && !"6C3E226B4D4795D518AB341B0824EC29".equals(input) // 'NULL'
//                && input.length() == 32) {
//            res = input;
//        }
//        return res;
//    }
//
//    //    private static JSONObject initPropJson(String propFile) throws IOException {
//    //        InputStream resourceAsStream =
//    // KfkResParse.class.getClassLoader().getResourceAsStream(propFile);
//    //        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
//    //        JSONObject res = new JSONObject();
//    //        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
//    //        while (keysIterator.hasNext()) {
//    //            String key = (String) keysIterator.next();
//    //            if(key.equals(CAPTIME)) {
//    //                res.put(key, -1);
//    //            } else {
//    //                res.put(key, "initV");
//    //            }
//    //        }
//    //        return res;
//    //    }
//
//    private static class WeKafkaKeyedSerializationSchema
//            implements KeyedSerializationSchema<String> {
//        @Override
//        public byte[] serializeKey(String element) {
//            JSONObject jsonObject = JSONObject.parseObject(element);
//            String keyby = jsonObject.getString(DEVICEID);
//            return keyby.getBytes();
//        }
//
//        @Override
//        public byte[] serializeValue(String element) {
//            return element.getBytes();
//        }
//
//        @Override
//        public String getTargetTopic(String element) {
//            return null;
//        }
//    }
//
//    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {
//
//        @Override
//        public int partition(
//                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
//            int partition = Math.abs(new String(key).hashCode() % partitions.length);
//            if (LOG.isDebugEnabled()) {
//                LOG.debug(
//                        " partitions: "
//                                + partitions.length
//                                + " partition: "
//                                + partition
//                                + " key: "
//                                + new String(key));
//            }
//            return partition;
//        }
//    }
//}
